package cn.itcast.streaming.function.map;

import cn.itcast.streaming.bean.ItcastDataPartObj;
import cn.itcast.streaming.bean.VehicelLocationInfo;
import cn.itcast.streaming.utils.GeoHashUtil;
import cn.itcast.streaming.utils.RedisUtil;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.functions.RichMapFunction;


//使用同步的方式拉宽地理位置数据
public class LocationInfoRedisFunction extends RichMapFunction<ItcastDataPartObj,ItcastDataPartObj> {


    //对原始数据流进行拉宽操作
    @Override
    public ItcastDataPartObj map(ItcastDataPartObj itcastDataPartObj) throws Exception {

        //住建 geoHash :经度+维度
        String geoHash = GeoHashUtil.encode(itcastDataPartObj.getLat(), itcastDataPartObj.getLng());
        //根据geohash获取redis的value值，
        byte[] locationInfo = RedisUtil.get(geoHash.getBytes());
        if(locationInfo!=null){
            //redis中存在该经纬度对应的地理位置信息
            VehicelLocationInfo vehicelLocationInfo = JSON.parseObject(new String(locationInfo), VehicelLocationInfo.class);
            if(vehicelLocationInfo!=null) {
                itcastDataPartObj.setCountry(vehicelLocationInfo.getCountry());
                itcastDataPartObj.setProvince(vehicelLocationInfo.getProvince());
                itcastDataPartObj.setCity(vehicelLocationInfo.getCity());
                itcastDataPartObj.setDistrict(vehicelLocationInfo.getDistrict());
                itcastDataPartObj.setAddress(vehicelLocationInfo.getAddress());
                //System.out.println("在redis中存在经纬度对应的地理位置信息。。。");
            }else {
                itcastDataPartObj.setCountry(null);
                itcastDataPartObj.setProvince(null);
                itcastDataPartObj.setCity(null);
                itcastDataPartObj.setDistrict(null);
                itcastDataPartObj.setAddress(null);
                //System.out.println("在redis中没有获取到经纬度对应的地理位置信息解析失败。。。");
            }
        }else {
            itcastDataPartObj.setCountry(null);
            itcastDataPartObj.setProvince(null);
            itcastDataPartObj.setCity(null);
            itcastDataPartObj.setDistrict(null);
            itcastDataPartObj.setAddress(null);
            //System.out.println("在redis中没有获取到经纬度对应的地理位置信息");
        }
        return itcastDataPartObj;
    }
}
